#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <iostream>
#include <queue>

using namespace std;

#define THREAD_NUM 5
//定义线程安全队列的类型

typedef void (*HANDLER)(int);
class QueType
{
    public:
        QueType()
        {

        }

        QueType(int data, HANDLER handler)
        {
            data_ = data;
            handler_ = handler;
        }

        void DealData()
        {
            handler_(data_);
        }
    private:
        //TODO data
        int data_;
        //deal function
        HANDLER handler_;
};

class ThreadPool
{
    public:
        ThreadPool()
        {
            capacity_ = 10;
            pthread_mutex_init(&que_lock_, NULL);
            pthread_cond_init(&cons_cond_, NULL);
            pthread_cond_init(&prod_cond_, NULL);
            thread_num_ = THREAD_NUM;

            is_quit_ = false;
        }

        ~ThreadPool()
        {
            pthread_mutex_destroy(&que_lock_);
            pthread_cond_destroy(&cons_cond_);
            pthread_cond_destroy(&prod_cond_);
        }


        int Start()
        {
            //1.create thread
            for(int i = 0; i < THREAD_NUM; i++)
            {
                int ret = pthread_create(&tid_[i], NULL, ThreadPollStart, (void*)this);
                if(ret < 0)
                {
                    perror("pthread_create");
                    thread_num_--;
                    continue;
                }
            }
            return thread_num_;
        }

        static void* ThreadPollStart(void* arg)
        {
            pthread_detach(pthread_self());
            ThreadPool* tp = (ThreadPool*)arg;

            while(1)
            {
                pthread_mutex_lock(&tp->que_lock_);
                while(tp->que_.empty())
                {
                    if(tp->is_quit_)
                    {
                        //pos
                        tp->thread_num_--;
                        pthread_mutex_unlock(&tp->que_lock_);
                        pthread_exit(NULL);
                    }
                    pthread_cond_wait(&tp->cons_cond_, &tp->que_lock_);
                }
                QueType qt;
                tp->Pop(&qt);
                pthread_mutex_unlock(&tp->que_lock_);

                pthread_cond_signal(&tp->prod_cond_);

                qt.DealData();
            }
            return NULL;
        }


        void Push(QueType& qt)
        {
            pthread_mutex_lock(&que_lock_);
            while(que_.size() >= capacity_)
            {
                pthread_cond_wait(&prod_cond_, &que_lock_);
            }
            que_.push(qt);
            pthread_mutex_unlock(&que_lock_);

            pthread_cond_signal(&cons_cond_);
        }

        void Pop(QueType* qt)
        {
            *qt = que_.front();
            que_.pop();
        }

        void exitThreadPool()
        {
            is_quit_ = true;
            while(thread_num_)
            {
                pthread_cond_signal(&cons_cond_);
            }
        }
    private:
        queue<QueType> que_;
        size_t capacity_;
        pthread_mutex_t que_lock_;
        pthread_cond_t cons_cond_;
        pthread_cond_t prod_cond_;

        pthread_t tid_[THREAD_NUM];
        int thread_num_;

        bool is_quit_;
};

void DeadFunc(int data)
{
    printf("data : %d\n", data);
}

int main()
{
    ThreadPool* tp = new ThreadPool();
    if(tp == NULL)
    {
        exit(1);
    }

    if(tp->Start() <= 0)
    {
        exit(2);
    }

    for(int i = 0; i < 100; i++)
    {
        QueType qt(i, DeadFunc);

        tp->Push(qt);
    }

    tp->exitThreadPool();

    delete tp;
    return 0;
}
